/**
 * Copyright (C), 2015-2018, XXX有限公司
 * FileName: DoubleAndTripleBoltApp
 * Author:   An-Il
 * Date:     2018/11/9 9:19
 * Description: wordcount实例
 * History:
 * <author>          <time>          <version>          <desc>
 * 作者姓名           修改时间           版本号              描述
 */
package com.blog.storm.example.integration.redis;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.redis.bolt.RedisStoreBolt;
import org.apache.storm.redis.common.config.JedisPoolConfig;
import org.apache.storm.redis.common.mapper.RedisDataTypeDescription;
import org.apache.storm.redis.common.mapper.RedisStoreMapper;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.ITuple;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;

import java.util.*;

/**
 * 使用storm实现wordcount实例,redis存储
 *
 * @url http://storm.apache.org/releases/1.2.2/storm-redis.html
 */
public class LocalWordCountRedisApp {

    /**
     * 数据源产生的Spout
     */
    public static class RandomWordSpout extends BaseRichSpout {

        SpoutOutputCollector collector;
        Random random;

        /**
         * 初始化
         *
         * @param conf      配置参数
         * @param context   上下文
         * @param collector 数据发射器
         */
        @Override
        public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
            this.collector = collector;
            this.random = new Random();
        }

        /**
         * 此方法是个死循环
         */
        @Override
        public void nextTuple() {
            Utils.sleep(100);
            String[] words = new String[]{"hadoop", "spark", "kafka", "flume", "storm", "hbase", "hive", "zookeeper"};
            String word = words[random.nextInt(words.length)];

            //将数据发送到下游
            this.collector.emit(new Values(word), UUID.randomUUID());
            System.out.println("send val:" + word);
        }

        /**
         * 申明输出字段
         *
         * @param declarer
         */
        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("word"));
        }

        @Override
        public void fail(Object msgId) {
            System.err.println(msgId + "处理失败");
        }
    }

    /**
     * 接受数据并处理
     */
    public static class WordCountBolt extends BaseRichBolt {

        OutputCollector collector;
        Map<String, Integer> countsMap;

        @Override
        public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
            this.collector = collector;
            countsMap = new HashMap<>();
        }

        @Override
        public void execute(Tuple tuple) {
            try {
                /**
                 * 这里通过上面定义的Fields进行获取
                 * 也可以通过下标获取：tuple.getInteger(0)
                 */
                String word = tuple.getStringByField("word");
                Integer count = countsMap.get(word);
                if (count == null) {
                    count = 0;
                }
                count++;

                //单词数汇总
                countsMap.put(word, count);

                //输出
                this.collector.emit(tuple, new Values(word, count));
                this.collector.ack(tuple);
            } catch (Exception e) {
                this.collector.fail(tuple);
            }
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("word", "count"));
        }
    }

    /**
     * 存储数据到redis
     */
    public static class WordCountStoreMapper implements RedisStoreMapper {
        private RedisDataTypeDescription description;
        private final String hashKey = "mystorm:wordCount";

        public WordCountStoreMapper() {
            description = new RedisDataTypeDescription(
                    RedisDataTypeDescription.RedisDataType.HASH, hashKey);
        }

        /**
         * 说明我们要使用的数据类型
         * @return
         */
        @Override
        public RedisDataTypeDescription getDataTypeDescription() {
            return description;
        }

        @Override
        public String getKeyFromTuple(ITuple tuple) {
            return tuple.getStringByField("word");
        }

        @Override
        public String getValueFromTuple(ITuple tuple) {
            return tuple.getIntegerByField("count").toString();
        }
    }


    /**
     * 构建本地Topology
     *
     * @param args
     * @url http://storm.apache.org/releases/1.2.2/Local-mode.html
     * http://storm.apache.org/releases/1.2.2/Running-topologies-on-a-production-cluster.html
     */
    public static void main(String[] args) {

        /**
         * 定义拓扑,在生产群集上运行拓扑与在本地模式下运行类似
         * Topology需要指定相关的Spout和Bolt的执行顺序
         * shuffleGrouping("dataSourceSpout"):Tuples以一种随机分布方式在Bolt的任务中,每个Bolt都保证获得相同数量的Tuples
         */
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("RandomWordSpout", new RandomWordSpout());
        builder.setBolt("WordCountBolt", new WordCountBolt()).shuffleGrouping("RandomWordSpout");

        //存储到redis的bolt
        JedisPoolConfig poolConfig = new JedisPoolConfig.Builder()
                .setHost("127.0.0.1").setPort(6379).build();
        RedisStoreMapper storeMapper = new WordCountStoreMapper();
        RedisStoreBolt storeBolt = new RedisStoreBolt(poolConfig, storeMapper);
        builder.setBolt("RedisStoreBolt", storeBolt).shuffleGrouping("WordCountBolt");


        StormTopology topology = builder.createTopology();
        //本地模式模拟正在进行的Storm集群，只需使用LocalCluster类
        LocalCluster cluster = new LocalCluster();
        /**
         * 然后，您可以使用submitTopology该LocalCluster对象上的方法提交拓扑。就像StormSubmitter上的相应方法一样，
         * submitTopology采用名称，拓扑配置和拓扑对象。然后，您可以使用killTopology将拓扑名称作为参数的方法来终止拓扑。
         */
        cluster.submitTopology("LocalWordCountRedisApp", new Config(), topology);

    }

}